/**
 * Copyright(c) 2014 Xunton Solutions
 *
 * History:
 *   15-12-7 上午10:37 Created by dwzhang
 */
package com.jade.modules.message.service.impl;

import java.util.Date;
import java.util.Map;
import java.util.UUID;
import javax.inject.Inject;
import javax.inject.Named;

import com.jade.framework.base.log.Log;
import com.jade.framework.base.log.LogUtils;
import com.jade.framework.base.pattern.pipeline.Pipeline;
import com.jade.framework.base.util.JsonUtils;
import com.jade.framework.mom.MomException;
import com.jade.framework.mom.MomService;
import com.jade.journal.Journal;
import com.jade.journal.JournalUtils;
import com.jade.modules.message.service.*;
import com.jade.modules.message.service.pipeline.MessageContext;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;

/**
 * 消息服务实现。
 *
 * @author <a href="mailto:dwzhang@xunton.com">dwzhang</a> 
 * @version 1.0 15-12-7 上午10:37
 */
@Named ("module_messageService")
public class MessageServiceImpl
        implements MessageService
{
    protected Log log = LogUtils.getLog(this.getClass());
    protected MomService momService;
    protected long[] delayLevels = {1 * 1000, 5 * 1000, 10 * 1000, 30 * 1000, 1 * 60 * 1000, 2 * 60 * 1000,
            3 * 60 * 1000, 4 * 60 * 1000, 5 * 60 * 1000, 6 * 60 * 1000, 7 * 60 * 1000,
            8 * 60 * 1000, 9 * 60 * 1000, 10 * 60 * 1000, 20 * 60 * 1000, 30 * 60 * 1000,
            60 * 60 * 1000, 120 * 60 * 1000};
    protected Map<String, Pipeline<MessageContext, MessageException>> pipelineMap;
    protected Map<String, MessageConfig> configMap;
    protected Map<String, MessageEventListener> listenerMap;
    private Journal journal = JournalUtils.getJournal(MessageJournal.NAME);

    public void setConfigMap(Map<String, MessageConfig> configMap)
    {
        this.configMap = configMap;
    }

    public void setPipelineMap(Map<String, Pipeline<MessageContext, MessageException>> pipelineMap)
    {
        this.pipelineMap = pipelineMap;
    }

    public void setListenerMap(Map<String, MessageEventListener> listenerMap)
    {
        this.listenerMap = listenerMap;
    }

    @Inject
    public void setMomService(@Named ("system_momService") MomService momService)
    {
        this.momService = momService;
    }

    protected Pipeline<MessageContext, MessageException> getPipeline(MessageContext context)
    {
        return pipelineMap == null ? null : pipelineMap.get(context.getType());
    }

    protected MessageConfig getConfig(Message message)
    {
        return configMap == null ? null : configMap.get(message.getType());
    }

    protected MessageEventListener getListener(Message message)
    {
        return listenerMap == null ? null : listenerMap.get(message.getType());
    }

    public void asyncProcess(Message message)
            throws MessageException
    {
        //generate message id
        String id = DigestUtils.md5Hex(UUID.randomUUID().toString());
        message.setId(id);
        enqueueMessage(message);
        journalMessage("accept", message, null);
    }

    public void syncProcess(Message message)
            throws MessageException
    {
        String id = DigestUtils.md5Hex(UUID.randomUUID().toString());
        message.setId(id);
        process(message, true);
    }

    protected void enqueueMessage(Message message)
            throws MessageException
    {
        int level = message.getScheduleTime() == null ? 0 : calculateDelayLevel(message.getScheduleTime());
        try {
            MessageConfig config = getConfig(message);
            String topic = (config == null || StringUtils.isBlank(
                    config.getTopic())) ? MessageConstants.DEFAULT_TOPIC : config.getTopic();
            momService.send(topic, message.getId(), message.getType(), level, message);
            log.info("enqueue message", message);
        }
        catch (MomException t) {
            log.warnThrowable("It's failed to enqueue message" + JsonUtils.writeToString(message), t);
            throw new MessageException("err.module.message.service.enqueue", t.getStatusCode() + ":" + t.getMessage(),
                    t);
        }
    }

    protected void process(Message message, boolean sync)
    {
        MessageContext context = new MessageContext(message);
        Pipeline<MessageContext, MessageException> pipeline = getPipeline(context);
        if (pipeline == null) {
            log.warn("Message pipeline not found", message);
        }
        else {
            log.debug("start to process message ", message);

            try {
                journalMessage("process", message, null);
                pipeline.handle(context);
                finishMessage(message, true, null);
                log.info("It's finished to process message ", message);
            }
            catch (MessageException e) {
                log.warnThrowable("It's failed to proces message " + message.getId(), e);
                if (!sync) {
                    //check if need to retry the message
                    MessageConfig config = getConfig(message);
                    int retryTimes = message.getIntParam(MessageConstants.PARAM_RETRY_TIMES, 0);
                    int maxRetryTimes = message.getMaxRetryTimes();
                    if (maxRetryTimes == 0 && config != null) {
                        maxRetryTimes = config.getMaxRetryTimes();
                    }
                    if (maxRetryTimes > 0 && retryTimes < maxRetryTimes) {
                        //retry the message
                        int interval = config == null ? MessageConstants.DEFAULT_INTERVAL : config.getInterval();
                        message.setScheduleTime(new Date(System.currentTimeMillis() + interval * 1000));
                        try {
                            enqueueMessage(message);
                        }
                        catch (MessageException me) {
                            // it's failed
                            finishMessage(message, false, "Can't retry message:" + e.getMessage());
                        }
                    }
                    else {
                        //it's faild to message
                        finishMessage(message, false, "Failed to process message:" + e.getMessage());
                    }
                }
            }
        }
    }

    protected void finishMessage(Message message, boolean result, String remark)
    {
        //触发处理事件
        MessageEventListener listener = getListener(message);
        if (listener != null) {
            try {
                if (result) {
                    listener.finished(message);
                }
                else {
                    listener.failed(message);
                }
            }
            catch (Throwable t) {
                log.warn("Can't handle message event:", t);
            }
        }
        //记录日志
        String action = result ? "finish" : "fail";
        journalMessage(action, message, remark);
    }

    protected void journalMessage(String action, Message message, String remark)
    {
        MessageJournal messageJournal = new MessageJournal();
        messageJournal.setId(message.getId());
        messageJournal.setType(message.getType());
        messageJournal.setMessageGroup(message.getGroup());
        messageJournal.setFid(message.getFid());
        messageJournal.setParams(JsonUtils.writeToString(message.getParams()));
        messageJournal.setRemark(remark);
        messageJournal.setAction(action);
        messageJournal.setActionDate(new Date());
        messageJournal.setScheduleDate(message.getScheduleTime());
        try {
            journal.infoEntity(messageJournal);
        }
        catch (Throwable t) {
            log.warn("Can't journal message:", t);
        }
    }

    protected int calculateDelayLevel(Date scheduleTime)
    {
        Date current = new Date();
        long n = scheduleTime.getTime() - current.getTime();
        int level = 0;
        if (n <= 1000) {
            return 0;
        }
        for (int i = delayLevels.length - 1; i >= 0; i--) {
            if (n >= delayLevels[i]) {
                level = i;
                break;
            }
        }
        return level + 1;
    }
}
